package nl.altenpts.net;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

/**
 * This class implements a reactor that builds upon on the new I/O java package (java.nio).
 * This class is thread safe.
 * 
 * @author fabian
 *
 */
public class NetReactor implements Runnable {

	/**
	 * 
	 * @throws IOException if Selector.open() fails
	 * 
	 * @see java.nio.channels.Selector
	 */
	public NetReactor() throws IOException {
			
	}
	
	/**
	 * Adds a task to reactor. The reactor will run task at the
	 * specified time. 
	 * 
	 * @param task The task to be executed. 
	 * @param when The Unix time in seconds when the task will be executed.
	 * @param threaded true if the task is to be executed in a thread of its own.
	 */
	
	public synchronized void addTask(Runnable task, long when, boolean threaded) {
		ScheduledTask timedtask = new ScheduledTask(when, task, threaded);
		synchronized(newtasks) {
			newtasks.add(timedtask);
		}
		selector.wakeup();
		notify();
	}
	
	/**
	 * 
	 * Adds a socket to the reactor. When the channel is ready for one or more of
	 * the requested operations the handler will be notified. 
	 * 
	 * @param channel The channel that will be checked for readiness.
	 * @param handler The handler that is notified when channel is ready for one or
	 *        more requested operations.
	 * @throws ClosedChannelException if channel is closed 
	 */
	
	public synchronized void addSocket(SelectableChannel channel, ChannelHandler handler) {
			synchronized(newsockets) {
				newsockets.add(new NewSocket(channel, handler));
			}
			selector.wakeup();
			notify();
	}
	
	
	/**
	 * Make this NetReactor reread the operations that the
	 * handler is interested in for channel. The operations
	 * for a channel are automatically reread after the
	 * handler handler has been notified of ready operations.
	 * 
	 * channel must already be registered with this NetReactor
	 * using addSocket().
	 * 
	 * @param channel
	 */
	public void updateChannelOps(SelectableChannel channel) {		
		
		synchronized(newsockets) {
			SelectionKey key = channel.keyFor(selector);
			if(key != null) { 
				ChannelHandler handler = (ChannelHandler) key.attachment();
				key.interestOps(handler.getRequestedOps(channel));
				selector.wakeup();
			}		
		}
	}
	
	
	/**
	 * The reactor's main loop. This method is a never-ending loop
	 * that checks sockets for readiness and tasks for execution.
	 * 
	 * Use stop() to this stop to break out of this loop.
	 * 
	 */
	public void run() {
		long timeout;
		keeprunning = true;
		
		while (true) {
			synchronized(this) {
				if(!keeprunning) {
					break;
				}
				if(tasks.isEmpty() && selector.keys().isEmpty()) {
					try {
						wait();
					} catch (InterruptedException e) {
						// ignore
					}
				}
			}
			
			if(!keeprunning) {
				break;
			}
			
			timeout = processTasks();
			
			try {
				selector.select(timeout);
			} catch (IOException e) {
				// just try again
			}
			
			if(!keeprunning) {
				break;
			}
			
			processSockets();	
		}					
	}
	
	
	/**
	 * Stop the run loop
	 * 
	 * @see NetReactor.run
	 */
	public synchronized void stop() {
		keeprunning = false;		
		selector.wakeup();
		notify();
	}
	
	/**
	 * 
	 * Executes all scheduled tasks that are due. And returns the interval in
	 * milliseconds to the next scheduled task.
	 * 
	 * @return Interval when the first upcoming task needs to be executed. 0 if
	 *         there are no scheduled tasks.
	 */
	private long processTasks() {
		//System.out.println("Processing tasks");
		long timeout;
		long curtime = System.currentTimeMillis();
		
		synchronized(newtasks) {
			tasks.addAll(newtasks);
			newtasks.clear();
		}
		
		if(tasks.isEmpty()) {
			timeout = 0;
		} else {
			ScheduledTask task = tasks.first();
			timeout = task.when() - curtime;			
			while(timeout <= 0) {
				tasks.remove(task);
				if(task.isThreaded()) {
					new Thread(task.getTask()).start();
				} else {
					task.getTask().run();
				}
				
				if(!tasks.isEmpty()) {
					task = tasks.first();
					timeout = task.when() - curtime;
				} else {
					timeout = 0; break;
				}
			}
		}
		
		return timeout;
	}
	
	/**
	 * Notify the associated handlers for all channels that are
	 * ready for one or more requested operation.
	 */
	private void processSockets() {
		//System.out.println("Processing channels");
		Set<SelectionKey> keys = selector.selectedKeys();
		Iterator<SelectionKey> iterator = keys.iterator();
		int ops;
		
		if(selector.keys().size() != 0) {

			while(iterator.hasNext()) {
				SelectionKey key = iterator.next();
				ChannelHandler handler = (ChannelHandler) key.attachment();				
				handler.onReady(key.readyOps(), key.channel());
				ops = handler.getRequestedOps(key.channel());
				if(ops != 0) {
					key.interestOps(ops);
				} else {
					key.cancel();					
				}
			}
		}
		
		synchronized(newsockets) {
			Iterator<NewSocket> socketiterator = newsockets.iterator();
			
			while(socketiterator.hasNext()) {
				NewSocket ns = socketiterator.next();
				try {
					ns.channel.register(selector, ns.handler.getRequestedOps(ns.channel) , ns.handler);
				} catch (ClosedChannelException e) {
					// ignore
				}				
			}			
			newsockets.clear();			
		}		
	}
	
	class NewSocket {
		public NewSocket(SelectableChannel channel, ChannelHandler handler) {
			this.channel = channel;
			this.handler = handler;
		}
		
		public SelectableChannel getChannel() {
			return channel;
		}
		
		public ChannelHandler getHandler() {
			return handler;
		}
		
		private SelectableChannel channel;
		private ChannelHandler handler;
		
	}
	
	/**
	 * Selector to check for ready channels.
	 */
	private Selector selector = Selector.open();
	/**
	 * New sockets
	 */
	private LinkedList<NewSocket> newsockets = new LinkedList<NewSocket>();
	
	/**
	 * New tasks
	 */
	private LinkedList<ScheduledTask> newtasks = new LinkedList<ScheduledTask>();
	
	/**
	 * Sorted set of scheduled tasks
	 */
	private SortedSet<ScheduledTask> tasks = new TreeSet<ScheduledTask>();
	/**
	 * keep running flag
	 */
	private boolean keeprunning;

}
